package com.wdl.market.app

import com.wdl.market.bean.{MarketUserBehavior, MarketViewCount}
import com.wdl.market.utils.SimulateMarketEventSource
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time


object MarketTotalApp {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val dataStream: DataStream[MarketUserBehavior] = env.addSource(new SimulateMarketEventSource)
      .assignAscendingTimestamps(_.timestamp)

    val resultStream: DataStream[MarketViewCount] = dataStream

      /** 过滤掉卸载行为 */
      .filter(_.behavior != "UNINSTALL")
      .map(data => ("total", 1L))
      /** 按照渠道和行为类型分组 */
      .keyBy(_._1)
      .timeWindow(Time.hours(1), Time.seconds(5))
      .aggregate(new MarketAgg, new MarketCountResult)


    resultStream.print("result:")

    env.execute()
  }
}
